/*
 *
 *  * Copyright (c) iwindplus Technologies Co., Ltd.2024-2030, All rights reserved.
 *
 *
 */

package com.iwindplus.base.util;

import cn.hutool.core.map.MapUtil;
import cn.hutool.core.util.PrimitiveArrayUtil;
import cn.hutool.core.util.StrUtil;
import com.iwindplus.base.domain.constant.CommonConstant;
import com.iwindplus.base.domain.constant.CommonConstant.NumberConstant;
import com.iwindplus.base.domain.constant.CommonConstant.ObservationConstant;
import com.iwindplus.base.domain.constant.CommonConstant.SymbolConstant;
import com.iwindplus.base.domain.vo.ResultVO;
import com.iwindplus.base.util.domain.dto.ReactorRequestDTO;
import com.iwindplus.base.util.domain.dto.ReactorResponseDTO;
import io.micrometer.observation.Observation;
import java.nio.charset.StandardCharsets;
import java.util.Map;
import java.util.Objects;
import java.util.function.Function;
import java.util.stream.Stream;
import org.reactivestreams.Publisher;
import org.springframework.core.io.buffer.DataBuffer;
import org.springframework.core.io.buffer.DataBufferUtils;
import org.springframework.http.HttpHeaders;
import org.springframework.http.HttpStatusCode;
import org.springframework.http.MediaType;
import org.springframework.http.server.reactive.ServerHttpRequest;
import org.springframework.http.server.reactive.ServerHttpRequestDecorator;
import org.springframework.http.server.reactive.ServerHttpResponse;
import org.springframework.http.server.reactive.ServerHttpResponseDecorator;
import org.springframework.web.server.ServerWebExchange;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;

/**
 * Reactor工具类.
 *
 * @author zengdegui
 * @since 2025/10/20 23:00
 */
public class ReactorUtil {

    private ReactorUtil() {
        throw new IllegalStateException(CommonConstant.UTILITY_CLASS);
    }

    /**
     * 请求体（缓存用）.
     */
    public static final String REQUEST_BODY = "requestBody";

    /**
     * 响应体（缓存用）
     */
    public static final String RESPONSE_BODY = "responseBody";

    /**
     * 获取响应结果.
     *
     * @param exchange       请求
     * @param httpStatusCode 状态码
     * @param data           数据
     * @return Mono<Void>
     */
    public static Mono<Void> getMonoResponse(ServerWebExchange exchange, HttpStatusCode httpStatusCode, ResultVO<Object> data) {
        ServerHttpResponse response = exchange.getResponse();
        response.setStatusCode(httpStatusCode);
        response.getHeaders().setContentType(MediaType.APPLICATION_JSON);

        return response.writeWith(Mono.fromCallable(() -> {
            byte[] bytes = JacksonUtil.toJsonBytes(data);
            return response.bufferFactory().wrap(bytes);
        }));
    }

    /**
     * 获取请求体.
     *
     * @param exchange 当前交换器
     * @return Mono<ReactorRequestDTO>
     */
    public static Mono<ReactorRequestDTO> getRequestBodyByAttr(ServerWebExchange exchange) {
        ReactorRequestDTO cached = exchange.getAttribute(REQUEST_BODY);
        return Mono.justOrEmpty(cached);
    }

    /**
     * 读取并缓存请求体，随后执行任意 Function<T,Mono<T>>，最终返回 Mono<ReactorRequestDTO>.
     *
     * @param exchange 当前交换器
     * @param function 要对“新” exchange 执行的逻辑
     * @param <T>      function 返回的 Mono 的泛型，可以任意
     * @return 携带 body 的 Mono<ReactorRequestDTO>
     */
    public static <T> Mono<ReactorRequestDTO> readRequestBody(ServerWebExchange exchange, Function<ServerWebExchange, Mono<T>> function) {
        final ServerHttpRequest request = exchange.getRequest();
        final HttpHeaders headers = request.getHeaders();
        final MediaType contentType = headers.getContentType();

        ReactorRequestDTO collector = new ReactorRequestDTO();
        collector.setRequestHeaders(headers.toSingleValueMap());
        collector.setQueryParams(request.getQueryParams().toSingleValueMap());

        boolean supported = Stream.of(MediaType.APPLICATION_JSON,
                MediaType.APPLICATION_FORM_URLENCODED)
            .anyMatch(mt -> mt.isCompatibleWith(contentType));
        if (!supported) {
            // 把 body 放到 attributes 里
            setAttribute(exchange, REQUEST_BODY, collector);
            return function.apply(exchange).then(Mono.just(collector));
        }

        return DataBufferUtils.join(request.getBody())
            .flatMap(aggregatedBuf -> {
                byte[] bytes = new byte[aggregatedBuf.readableByteCount()];
                aggregatedBuf.read(bytes);
                DataBufferUtils.release(aggregatedBuf);

                // 1. 构造新请求（可重复读）
                ServerHttpRequest newRequest = ReactorUtil.buildNewServerHttpRequest(exchange, bytes);
                ServerWebExchange newExchange = exchange.mutate().request(newRequest).build();

                collector.setRequestBody(new String(bytes));
                // 2. 把 body 放到 attributes 里
                setAttribute(newExchange, REQUEST_BODY, collector);

                // 3. 执行业务函数
                return function.apply(newExchange).then(Mono.just(collector));
            }).doOnDiscard(DataBuffer.class, DataBufferUtils::release);
    }

    /**
     * 获取响应体.
     *
     * @param exchange 当前交换器
     * @return Mono<ReactorResponseDTO>
     */
    public static Mono<ReactorResponseDTO> getResponseBodyByAttr(ServerWebExchange exchange) {
        ReactorResponseDTO cached = exchange.getAttribute(RESPONSE_BODY);
        return Mono.justOrEmpty(cached);
    }

    /**
     * 读取并缓存响应体，随后执行任意 Function<T,Mono<T>>，最终返回 Mono<ReactorResponseDTO>.
     *
     * @param exchange 当前交换器
     * @param function 要对“新” exchange 执行的逻辑
     * @param <T>      function 返回的 Mono 的泛型
     * @return 携带响应体的 Mono<ReactorResponseDTO>
     */
    public static <T> Mono<ReactorResponseDTO> readResponseBody(ServerWebExchange exchange, Function<ServerWebExchange, Mono<T>> function) {
        // 提前分配一个容器，用来聚合下游写出的数据
        ReactorResponseDTO collector = new ReactorResponseDTO();

        // 1. 构造可重复读的响应
        ServerHttpResponseDecorator newResp = ReactorUtil.buildNewResponseDecorator(exchange, collector);
        ServerWebExchange newExchange = exchange.mutate().response(newResp).build();

        // 2. 执行业务逻辑
        return function.apply(newExchange).then(Mono.just(collector));
    }

    /**
     * 构建新请求（重新包装请求体）.
     *
     * @param exchange  请求
     * @param bodyBytes 请求体字节数组
     * @return ServerHttpRequest
     */
    public static ServerHttpRequest buildNewServerHttpRequest(ServerWebExchange exchange, byte[] bodyBytes) {
        final DataBuffer newBuffer = exchange.getResponse().bufferFactory().wrap(bodyBytes);
        // 重新包装请求体
        return new ServerHttpRequestDecorator(exchange.getRequest()) {
            @Override
            public Flux<DataBuffer> getBody() {
                return Flux.just(newBuffer);
            }
        };
    }

    /**
     * 构建新响应（重新包装响应体）.
     *
     * @param exchange  当前交换器
     * @param collector 数据收集器
     * @return ServerHttpResponseDecorator
     */
    public static ServerHttpResponseDecorator buildNewResponseDecorator(ServerWebExchange exchange, ReactorResponseDTO collector) {
        ServerHttpResponse originalResp = exchange.getResponse();
        return new ServerHttpResponseDecorator(originalResp) {
            @Override
            public Mono<Void> writeWith(Publisher<? extends DataBuffer> body) {
                return DataBufferUtils.join(body)
                    .flatMap(buf -> {
                        final byte[] bodyBytes = HttpsUtil.getBytes(buf);
                        fillResponse(exchange, bodyBytes);
                        DataBuffer newBuf = bufferFactory().wrap(bodyBytes);
                        return super.writeWith(Flux.just(newBuf));
                    });
            }

            private void fillResponse(ServerWebExchange exchange, byte[] bytes) {
                // 响应头
                final Map<String, String> headers = getDelegate().getHeaders().toSingleValueMap();
                collector.setResponseHeaders(headers);
                // 响应状态码
                collector.setResponseStatus(getStatusCode() != null ? getStatusCode().value() : null);
                // 响应体
                collector.setResponseBody(new String(bytes));

                setAttribute(exchange, RESPONSE_BODY, collector);
            }
        };
    }

    /**
     * 截取数据.
     *
     * @param bytes 字节数组
     * @param limit 限制长度（可选）
     * @return String
     */
    public static String getLimitByBytes(byte[] bytes, Integer limit) {
        if (PrimitiveArrayUtil.isEmpty(bytes)) {
            return null;
        }
        String str = new String(bytes);
        if (Objects.nonNull(limit)) {
            return StrUtil.maxLength(str, limit * NumberConstant.NUMBER_ONE_THOUSAND_TWENTY_FOUR);
        }
        return str;
    }

    /**
     * 构建查询参数.
     *
     * @param request 请求
     * @return String
     */
    public static String buildQueryParam(ServerHttpRequest request) {
        Map<String, String> singleMap = request.getQueryParams().toSingleValueMap();
        return MapUtil.isEmpty(singleMap) ? SymbolConstant.EMPTY_STR : UrlUtil.buildQuery(singleMap, StandardCharsets.UTF_8);
    }

    /**
     * 设置属性.
     *
     * @param exchange 请求
     * @param key      键
     * @param value    值
     */
    public static <T> void setAttribute(ServerWebExchange exchange, String key, T value) {
        exchange.getAttributes().put(key, value);
    }

    /**
     * 获取属性.
     *
     * @param exchange 请求
     * @param key      键
     * @return T
     */
    public static <T> T getAttribute(ServerWebExchange exchange, String key) {
        return exchange.getAttribute(key);
    }

    /**
     * 构建Observation.
     *
     * @param exchange   请求
     * @param metricName 指标名称
     * @return Mono<Void>
     */
    public static Mono<Observation> buildObservation(ServerWebExchange exchange, String metricName) {
        return Mono.defer(() -> {
            final ServerHttpRequest request = exchange.getRequest();
            final ServerHttpResponse response = exchange.getResponse();
            Map<String, String> tags = Map.of(
                ObservationConstant.HTTP_METHOD, request.getMethod().name(),
                ObservationConstant.HTTP_URI, request.getPath().value(),
                ObservationConstant.HTTP_STATUS, String.valueOf(response.getStatusCode().value()));
            return Mono.just(ObservationUtil.of(metricName, tags));
        });
    }

}
